package com.cqs.learning.spring.cloud.app.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.cqs.learning.spring.cloud.app.bean.dto.HousePriceChannel;
import com.cqs.learning.spring.cloud.app.bean.po.HousePrice;
import com.cqs.learning.spring.cloud.app.service.HousePriceService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;

import javax.annotation.Resource;
import java.util.Random;
import java.util.concurrent.*;

/**
 * @author li
 * @Date 5/31/20 11:25 PM
 */
@EnableBinding(value = HousePriceChannel.class)
@Slf4j
public class HousePriceStreamConsumer implements InitializingBean {


    @Resource
    HousePriceService housePriceService;

    private final static int MAX_CAPACITY = 10000;
    public final static BlockingQueue<Runnable> housePriceQueue = new ArrayBlockingQueue<Runnable>(MAX_CAPACITY);

    private ThreadPoolExecutor executor = new ThreadPoolExecutor(24, 32, 60, TimeUnit.SECONDS,
            housePriceQueue, r -> new Thread(r, "house-price-"), new ThreadPoolExecutor.AbortPolicy());
    private Random rand = new Random();

    /**
     * @param message
     */
    @StreamListener(HousePriceChannel.INPUT)
    public void consumerHousePriceMsg(Message<String> message) {
        String payload = message.getPayload();
        //log.info("订阅House Price消息：{},housePriceService:{}", payload, housePriceService);
        housePriceQueue.offer(() -> subTask(payload));
        if (housePriceQueue.size() > (MAX_CAPACITY * 0.75) ) {
            log.warn("队列中任务数量过多,进入休眠状态");
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }



    public void subTask(String payload) {
        //1. 转化为对象
        HousePrice housePrice = JSON.parseObject(payload, HousePrice.class);
        JSONObject jo = JSON.parseObject(payload);
        housePrice.setTui(jo.getString("Transaction unique identifier"));
        long start = System.currentTimeMillis();
        boolean save = housePriceService.save(housePrice);
        long cost = System.currentTimeMillis() - start;
        if (!save) {
            log.warn("保存失败:{},cost:{}",  payload,cost );
        }else {
            if (rand.nextInt(100) < 10){
                log.info("保存成功:id{},cost:{}", housePrice.getId(),cost);
            }
        }
    }

    /**
     * 消费消息
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet()  {

        Runnable r = () -> {
            while (true){
                try {
                    Runnable poll = housePriceQueue.take();
                    executor.submit(poll);
                    if (rand.nextInt(100) < 6){
                        long taskCount = executor.getTaskCount();
                        long queueSize = executor.getQueue().size();
                        long completedTaskCount = executor.getCompletedTaskCount();
                        log.info("队列的任务taskCount:{}:size:{},complete task:{}" ,taskCount,
                                queueSize,completedTaskCount);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        new Thread(r,"kafka-task-submit").start();
    }

    public static void main(String[] args) {
        String s = "{\"PropertyType\":\"F\",\"District\":\"TORBAY\",\"@version\":\"1\",\"PAON\":\"VILLA PARADISO\",\"type\":\"csv_log\",\"path\":\"/home/li/workspace/dataset/HousePriceDataEnglandWales1995to2014/pp-1995.csv/pp-1995.csv\",\"Transaction unique identifier\":\"{A42E2F04-2538-4A25-94C5-49E29C6C8FA8}\",\"Duration\":\"L\",\"SAON\":\"FLAT 10\",\"TransferDate\":\"1995-01-31 00:00\",\"ppdCategory\":\"A\",\"Street\":\"HIGHER WARBERRY ROAD\",\"RecordStatus\":\"A\",\"OldOrNew\":\"N\",\"Locality\":\"TORQUAY\",\"host\":\"cqs\",\"@timestamp\":\"2020-06-01T16:25:18.379Z\",\"Price\":\"18500\",\"TownOrCity\":\"TORQUAY\",\"County\":\"TORBAY\",\"message\":\"\\\"{A42E2F04-2538-4A25-94C5-49E29C6C8FA8}\\\",\\\"18500\\\",\\\"1995-01-31 00:00\\\",\\\"TQ1 1RY\\\",\\\"F\\\",\\\"N\\\",\\\"L\\\",\\\"VILLA PARADISO\\\",\\\"FLAT 10\\\",\\\"HIGHER WARBERRY ROAD\\\",\\\"TORQUAY\\\",\\\"TORQUAY\\\",\\\"TORBAY\\\",\\\"TORBAY\\\",\\\"A\\\",\\\"A\\\"\\r\",\"Postcode\":\"TQ1 1RY\"}";
        JSONObject jsonObject = JSON.parseObject(s);
        HousePrice housePrice = JSON.parseObject(s, HousePrice.class);
        housePrice.setTui(jsonObject.getString("Transaction unique identifier"));
        System.out.println(JSON.toJSON(housePrice));
    }

}